package com.hexinfo.dmpro.xxljob.service.jobhandler;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.async.AsyncTableTask;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibraryTask;
import com.hexinfo.dmpro.sparing.service.*;
import com.xqfunds.job.core.biz.model.ReturnT;
import com.xqfunds.job.core.handler.annotation.XxlJob;
import com.xqfunds.job.core.log.XxlJobLogger;
//import com.xxl.job.core.biz.model.ReturnT;
//import com.xxl.job.core.context.XxlJobHelper;
//import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * XxlJob开发示例（Bean模式）
 *
 * 开发步骤：
 *      1、任务开发：在Spring Bean实例中，开发Job方法；
 *      2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
 *      3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
 *      4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；
 *
 * @author xuxueli 2019-12-11 21:52:51
 */
@Component
public class SampleXxlJob {
    private static final Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);

    @Autowired
    private SyncRuleConfService syncRuleConfService;
    @Autowired
    private SyncRuleConfDatabaseService syncRuleConfDatabaseService;
    @Autowired
    private CreateTableService createTableService;
    @Autowired
    private AsyncTableTask asyncTableTask;
    @Autowired
    private FailTableRetryService failTableRetryService;
    @Autowired
    private SyncCompleteLibraryTaskService syncCompleteLibraryTaskService;
    @Autowired
    private SyncCompleteLibraryService syncCompleteLibraryService;
    @Autowired
    private LibraryLevelRetryService libraryLevelRetryService;
    /**
     * 调度调用集群热备
     */
    @XxlJob("heatBackup")
    public ReturnT<String> heatBackup(String param) throws Exception {
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup begin");
        Map<String, String> map = new ObjectMapper().readValue(param, new TypeReference<Map<String, String>>() {});
        logger.info("Automatic heat backup | 自动热备参数 | param：{}", map.toString());

        //大表同步（只保留两年内日期分区数据）
        asyncTableTask.tableSync((String) map.get("source"), "", "");

        //热备逻辑
        Boolean aBoolean = syncRuleConfService.qyScanMetadataWhere(
                (String) map.get("flag"),
                (String) map.get("source"),
                (String) map.get("cpWhether"));

        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup end");
        if (aBoolean){
            return ReturnT.SUCCESS;
        }else {
            return ReturnT.FAIL;
        }
    }

    /**
     * 热备失败重跑
     */
    @XxlJob("failTableReRun")
    public ReturnT<String> failTableReRun(String param) throws Exception {
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup begin");
        Map<String, String> map = new ObjectMapper().readValue(param, new TypeReference<Map<String, String>>() {});
        logger.info("Automatic heat backup | 自动热备参数 | param：{}", map.toString());

        Boolean aBoolean = failTableRetryService.failTableRetrySource((String) map.get("source"));

        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup end");
        if (aBoolean){
            return ReturnT.SUCCESS;
        }else {
            return ReturnT.FAIL;
        }
    }

    /**
     * 全库数据热备同步
     */
    @XxlJob("syncRuleDatabaseData")
    public ReturnT<String> syncRuleDatabaseData(String param) throws Exception {
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup begin");
        Map<String, String> map = new ObjectMapper().readValue(param, new TypeReference<Map<String, String>>() {});
        logger.info("Automatic heat backup | 全库数据热备同步参数 | param：{}", map.toString());

        //同步
        Map<String, String> database = syncRuleConfDatabaseService.getSyncRuleDatabaseData((String) map.get("database"));
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup end");
        if (ObjectUtil.isNotEmpty(database)){
            com.hexinfo.dmpro.sparing.util.Logger.log("同步返回值："+database.toString());
            if (StrUtil.isNotBlank(database.get("success"))){
                return ReturnT.SUCCESS;
            }else {
                return ReturnT.FAIL;
            }
        }else {
            return ReturnT.FAIL;
        }
    }


    /**
     * 补充自动化建表
     */
    @XxlJob("createTableDifference")
    public ReturnT<String> createTableDifference(String param) throws Exception {
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup begin");
        Map<String, String> map = new ObjectMapper().readValue(param, new TypeReference<Map<String, String>>() {});
        logger.info("Automatic heat backup | 补充自动化建表参数 | param：{}", map.toString());

        //自动化建表
        String database = createTableService.createTableDifference();
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup end");
        if (ObjectUtil.isNotEmpty(database)){
            com.hexinfo.dmpro.sparing.util.Logger.log("同步返回值："+database);
            if (StrUtil.isNotBlank(database)){
                return ReturnT.SUCCESS;
            }else {
                return ReturnT.FAIL;
            }
        }else {
            return ReturnT.FAIL;
        }
    }

    /**
     * 库级重试
     */
    @XxlJob("levelRetry")
    public ReturnT<String> levelRetry(String param) throws Exception {
        com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup begin");
        Map<String, String> map = new ObjectMapper().readValue(param, new TypeReference<Map<String, String>>() {});
        logger.info("Automatic heat backup | 库级重试参数 | param：{}", map.toString());
        //逻辑
        String taskName;
        if(StrUtil.isNotBlank(map.get("taskName"))&&"${bizDate}".equals(map.get("taskName"))){
            taskName = map.get("bizDate");
        }else if(StrUtil.isBlank(map.get("taskName"))){
            taskName = DateUtil.format(DateUtil.date(), "yyyyMMdd");
        }else{
            taskName = map.get("taskName");
        }
        //根据任务名称和日期查询任务
        LambdaQueryWrapper<SyncCompleteLibraryTask> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(SyncCompleteLibraryTask::getTaskName,taskName)
                .eq(SyncCompleteLibraryTask::getDatabaseName,com.hexinfo.dmpro.common.utils.StringUtils.toLowerCase(map.get("database")))
                .eq(SyncCompleteLibraryTask::getDel, CommonConstants.STATUS_NORMAL);
        SyncCompleteLibraryTask task = syncCompleteLibraryTaskService.getBaseMapper().selectOne(wrapper);
        //根据任务ID查询失败表级任务
        LambdaQueryWrapper<SyncCompleteLibrary> wrapperLib = Wrappers.lambdaQuery();
        wrapperLib.eq(SyncCompleteLibrary::getTaskId,task.getId())
                .eq(SyncCompleteLibrary::getTaskState, ScanCommonConstants.ClusterName.SB.value)
                //排除自动建表任务和同步任务
                .notLike(SyncCompleteLibrary::getLibraryTableName,"【")
                .eq(SyncCompleteLibrary::getDel, CommonConstants.STATUS_NORMAL);
        List<SyncCompleteLibrary> list = syncCompleteLibraryService.getBaseMapper().selectList(wrapperLib);
        if(ObjectUtil.isNotEmpty(list)){
            //库级重试
            String database = libraryLevelRetryService.levelRetry(list);
            com.hexinfo.dmpro.sparing.util.Logger.log("Automatic heat backup end");
            if (ObjectUtil.isNotEmpty(database)){
                com.hexinfo.dmpro.sparing.util.Logger.log("同步返回值："+database);
                if ("success".equals(database)){
                    return ReturnT.SUCCESS;
                }else {
                    return ReturnT.FAIL;
                }
            }else {
                return ReturnT.FAIL;
            }
        }else{
            //无失败任务
            return ReturnT.SUCCESS;
        }
    }


    /**
     * 1、简单任务示例（Bean模式）
     */
    @XxlJob("demoJobHandler")
    public ReturnT<String> demoJobHandler(String param) throws Exception {
//        String jobParam = XxlJobHelper.getJobParam();
        //XxlJobLogger.log("XXL-JOB, Hello World.");
        logger.info(param);
        for (int i = 0; i < 3; i++) {
            //XxlJobLogger.log("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        return ReturnT.SUCCESS;
    }

    @XxlJob("returnDataJobHandler")
    public ReturnT<String> returnDataJobHandler(String param) throws Exception {
        logger.info(param);

        HashMap<String, String> map = new HashMap<String, String>();
        map.put("fdgdhfhgfgh", "ghfgh");

        ObjectMapper objectMapper = new ObjectMapper();
        String s = objectMapper.writeValueAsString(map);
        ReturnT<String> mapReturnT = new ReturnT<String>(s);

        return mapReturnT;
    }

    @XxlJob("returnSliceDataJobHandler")
    public ReturnT<String> returnSliceDataJobHandler(String param) throws Exception {
        List<Integer> collect = IntStream.range(0, 10).boxed().collect(Collectors.toList());

        ObjectMapper mapper = new ObjectMapper();
        String sliceStr = mapper.writeValueAsString(collect);

        ReturnT<String> mapReturnT = new ReturnT<String>(sliceStr);
        return mapReturnT;
    }

    /**
     * 4、跨平台Http任务
     */
    @XxlJob("httpJobHandler")
    public ReturnT<String> httpJobHandler(String param) throws Exception {

        // request
        HttpURLConnection connection = null;
        BufferedReader bufferedReader = null;
        try {
            // connection
            URL realUrl = new URL(param);
            connection = (HttpURLConnection) realUrl.openConnection();

            // connection setting
            connection.setRequestMethod("GET");
            connection.setDoOutput(true);
            connection.setDoInput(true);
            connection.setUseCaches(false);
            connection.setReadTimeout(5 * 1000);
            connection.setConnectTimeout(3 * 1000);
            connection.setRequestProperty("connection", "Keep-Alive");
            connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
            connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");

            // do connection
            connection.connect();

            //Map<String, List<String>> map = connection.getHeaderFields();

            // valid StatusCode
            int statusCode = connection.getResponseCode();
            if (statusCode != 200) {
                throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");
            }

            // result
            bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
            StringBuilder result = new StringBuilder();
            String line;
            while ((line = bufferedReader.readLine()) != null) {
                result.append(line);
            }
            String responseMsg = result.toString();

            //XxlJobLogger.log(responseMsg);
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            //XxlJobLogger.log(e);
            return ReturnT.FAIL;
        } finally {
            try {
                if (bufferedReader != null) {
                    bufferedReader.close();
                }
                if (connection != null) {
                    connection.disconnect();
                }
            } catch (Exception e2) {
                //XxlJobLogger.log(e2);
            }
        }

    }

    /**
     * 5、生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑；
     */
    @XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy")
    public ReturnT<String> demoJobHandler2(String param) throws Exception {
        //XxlJobLogger.log("XXL-JOB, Hello World.");
        logger.error("liml demoJobHandler2 param:" + param);
        return ReturnT.SUCCESS;
    }

    /**
     * 6、分片测试
     */
    @XxlJob(value = "sliceTest", init = "init", destroy = "destroy")
    public ReturnT<String> sliceTest(String param) throws Exception {

        //XxlJobLogger.log("sliceTest success");
        logger.error("liml sliceTest param:" + param);
        ReturnT<String> mapReturnT = new ReturnT<>("[\"a\", \"b\", \"c\", \"d\"]");
        return mapReturnT;
    }

    /**
     * 7、分片处理
     */
    @XxlJob(value = "sliceHandler")
    public ReturnT<String> sliceHandler(String param) throws Exception {
        ReturnT<String> mapReturnT = new ReturnT<>("sliceHandler success");
        return mapReturnT;
    }

    @XxlJob(value = "mergeHandler")
    public ReturnT<String> mergeHandler(String param) throws Exception {
        ReturnT<String> mapReturnT = new ReturnT<>("mergeHandler success");
        logger.error("liml mergeHandler param:" + param);
        return mapReturnT;
    }

    public void init() {
        logger.info("init");
    }

    public void destroy() {
        logger.info("destroy");
    }

}
